package com.apps.sdses.flink141.jar;

import com.apps.sdses.flink141.entity.Event;
import com.apps.sdses.flink141.source.KafkaEventDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaDeserializationSchemaWrapper;

/**
 * author: codejiwei
 * date: 2023/9/3
 * desc:
 **/
public class KafkaSourceTest {
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);


        KafkaDeserializationSchemaWrapper<String> schemaWrapper = new KafkaDeserializationSchemaWrapper<String>(new SimpleStringSchema());


//        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        KafkaSource<Event> kafkaSource = KafkaSource.<Event>builder()
                .setBootstrapServers("192.168.102.154:29092")
                .setTopics("drity_data_source")
                .setGroupId("test_jiwei_01")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setDeserializer(KafkaRecordDeserializationSchema.of(new KafkaEventDeserializationSchema()))
//                .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
//                .setDeserializer(KafkaRecordDeserializationSchema.of(schemaWrapper))
                .build();


        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source")
                .print();

        env.execute();

    }
}
